package com.xinqing.bigdata.flink.datastream.function;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author:CHQ
 * @Date:2021/5/24 15:18
 * @Description
 */
public class AvgAggreProcessFunction extends ProcessWindowFunction<Double, Tuple2<String, Double>, String, TimeWindow> {
    Logger logger = LoggerFactory.getLogger(AvgAggreProcessFunction.class);

    @Override
    public void process(String key, Context context, Iterable<Double> elements, Collector<Tuple2<String, Double>> out) throws Exception {
        logger.info("窗口开始时间为{}", context.window().getStart());
        for (Double element : elements) {
            Tuple2<String, Double> tuple2 = new Tuple2<>(key, element);
            logger.info("输出元素为->{}", tuple2.toString());
            out.collect(new Tuple2<>(key, element));
        }
        logger.info("窗口关闭时间为{}", context.window().getEnd());
    }
}
